Add Kafka event sourcing middleware by aliok · Pull Request #185 · knative-extensions/func-go

aliok · 2026-07-16T11:57:47Z

Summary

Add Kafka consumer middleware that converts Kafka messages to CloudEvents
and delivers them to the standard CloudEvents handler — no separate handler
signature needed
Kafka messages with ce_specversion headers (CloudEvents Kafka Protocol
Binding, binary content mode) are passed through as-is
Plain Kafka messages are wrapped in a CloudEvent with type
dev.knative.kafka.event and extensions for topic, partition, offset, key
Health-only HTTP server runs alongside for readiness/liveness probes
Exports GetReceiverFn from cloudevents package so kafka can reuse the
12 supported handler signatures without duplication

How it works

The kafka.Start(handler) function accepts any CloudEvents handler (same
signatures as cloudevents.Start). The scaffolding main.go branches at
runtime based on FUNC_TRANSPORT=kafka env var:

if os.Getenv("FUNC_TRANSPORT") == "kafka" {
    err = kafka.Start(f.New())
} else {
    err = ce.Start(f.New())
}

Configuration is via environment variables set by the func CLI deployer
from run.kafka in func.yaml:

KAFKA_BROKERS — comma-separated broker addresses
KAFKA_TOPIC — topic to consume from
KAFKA_CONSUMER_GROUP — consumer group ID

Delivery semantics

At-least-once, in-order per partition
Offset marked on successful handler return
Errors logged and skipped (not retried)

Limitations

Go only
No SASL/TLS authentication
No retry or dead-letter queue
Response events from handler are logged and ignored

Supersedes #169

Closes #183

Testing Kafka Functions End-to-End on Kind

This guide walks through testing the Kafka middleware on a local Kind cluster.

How it works

A CloudEvents function (invoke: cloudevent) can consume from Kafka by adding
a run.kafka section to func.yaml. The runtime automatically converts Kafka
messages to CloudEvents and delivers them to the same Handle method — no code
changes required.

The deployer/runner sets FUNC_TRANSPORT=kafka and KAFKA_BROKERS,
KAFKA_TOPIC, KAFKA_CONSUMER_GROUP env vars automatically from the
run.kafka config. The user never needs to set these manually.

Limitations

Go only: Kafka consumption is currently supported only for Go functions.

Testing

Prerequisites

kind
kubectl
Go 1.25+
Docker

1. Build the func CLI

cd ~/go/src/knative.dev/func
go build -o /tmp/func-local ./cmd/func

2. Create a Kind cluster with Knative

kind create cluster --name kafka-test

# Install Knative Serving
kubectl apply -f https://github.com/knative/serving/releases/latest/download/serving-crds.yaml
kubectl apply -f https://github.com/knative/serving/releases/latest/download/serving-core.yaml

# Install Kourier (networking layer)
kubectl apply -f https://github.com/knative/net-kourier/releases/latest/download/kourier.yaml
kubectl patch configmap/config-network \
  --namespace knative-serving \
  --type merge \
  --patch '{"data":{"ingress-class":"kourier.ingress.sigs.k8s.io"}}'

# Wait for Knative to be ready
kubectl wait --for=condition=Ready pods --all -n knative-serving --timeout=120s

3. Install Kafka using Strimzi

kubectl create namespace kafka
kubectl apply -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka
kubectl wait --for=condition=Ready pods --all -n kafka --timeout=120s

# Create a single-node Kafka cluster
kubectl apply -n kafka -f - <<EOF
apiVersion: kafka.strimzi.io/v1
kind: KafkaNodePool
metadata:
  name: dual-role
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 1
  roles:
    - controller
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 1Gi
        deleteClaim: true
---
apiVersion: kafka.strimzi.io/v1
kind: Kafka
metadata:
  name: my-cluster
  annotations:
    strimzi.io/node-pools: enabled
    strimzi.io/kraft: enabled
spec:
  kafka:
    version: 4.2.0
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
  entityOperator:
    topicOperator: {}
EOF

# Wait for Kafka to be ready (takes a few minutes)
kubectl wait kafka/my-cluster --for=condition=Ready --timeout=300s -n kafka

4. Create a Kafka topic

kubectl apply -n kafka -f - <<EOF
apiVersion: kafka.strimzi.io/v1
kind: KafkaTopic
metadata:
  name: test-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
EOF

5. Create the function

mkdir /tmp/my-kafka-func && cd /tmp/my-kafka-func
/tmp/func-local create -l go -t cloudevents

This creates a standard CloudEvents function. The generated function.go
has a Handle(e event.Event) (*event.Event, error) method — this same
handler will receive Kafka messages wrapped as CloudEvents.

The default template handler doesn't log the incoming event. To see messages
in the logs, add a print statement to function.go:

func (f *MyFunction) Handle(e event.Event) (*event.Event, error) {
	fmt.Printf("Received CloudEvent: %s\n", e)

	ret := event.New()
	// ... rest of handler ...

6. Add Kafka config

Edit func.yaml and add the run.kafka section:

# func.yaml
specVersion: 0.36.0
name: my-kafka-func
runtime: go
created: ...
invoke: cloudevent
run:
  kafka:
    brokers: "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092"
    topic: "test-topic"
    consumerGroup: "my-kafka-func-group"

That's all the user needs to do. The deployer translates this to env vars
(FUNC_TRANSPORT=kafka, KAFKA_BROKERS, KAFKA_TOPIC,
KAFKA_CONSUMER_GROUP) on the Knative Service automatically.

7. Build and deploy

Note: Until knative.dev/func-go is released with the kafka package,
func deploy can't resolve the module. Use the manual build below.
Once published, just run:
FUNC_REGISTRY=ttl.sh/my-kafka-func-test /tmp/func-local deploy --build --verbose

Manual build (pre-release)

# 1. Trigger scaffolding (will fail at compile — that's OK)
FUNC_REGISTRY=ttl.sh /tmp/func-local build --builder=host --verbose 2>&1 || true

# 2. Patch go.mod to use local func-go source
cd /tmp/my-kafka-func/.func/build
go mod edit -replace knative.dev/func-go=$HOME/go/src/knative.dev/func-go
go mod tidy

# 3. Cross-compile
GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o result/f .

# 4. Build and push container image
mkdir -p /tmp/kafka-func-build
cp result/f /tmp/kafka-func-build/f
cat > /tmp/kafka-func-build/Dockerfile <<'DOCKERFILE'
FROM gcr.io/distroless/static:nonroot
COPY f /usr/local/bin/f
ENTRYPOINT ["/usr/local/bin/f"]
DOCKERFILE

TTL_TAG="ttl.sh/my-kafka-func-$(date +%s):2h"
docker build --platform linux/amd64 -t "$TTL_TAG" /tmp/kafka-func-build
docker push "$TTL_TAG"
echo "Image: $TTL_TAG"

8. Deploy the Knative Service

kubectl apply -f - <<EOF
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: my-kafka-func
spec:
  template:
    metadata:
      annotations:
        autoscaling.knative.dev/min-scale: "1"
    spec:
      containers:
        - image: $TTL_TAG
          env:
            - name: FUNC_TRANSPORT
              value: "kafka"
            - name: KAFKA_BROKERS
              value: "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092"
            - name: KAFKA_TOPIC
              value: "test-topic"
            - name: KAFKA_CONSUMER_GROUP
              value: "my-kafka-func-group"
EOF

The min-scale: "1" annotation keeps the pod running — Kafka consumers must
stay up to receive messages.

Wait for the pod:

kubectl wait pods -l serving.knative.dev/service=my-kafka-func --for=condition=Ready --timeout=120s

9. Tail function logs

In a separate terminal:

kubectl logs -l serving.knative.dev/service=my-kafka-func -c user-container -f

You should see the consumer starting and connecting to Kafka.

10. Send test messages

Plain message (wrapped as CloudEvent)

kubectl run kafka-producer -n kafka \
  --image=quay.io/strimzi/kafka:latest-kafka-4.2.0 \
  --restart=Never \
  --command -- sh -c \
  'echo "Hello from Kafka!" | bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic test-topic'

kubectl wait pod/kafka-producer -n kafka --for=jsonpath='{.status.phase}'=Succeeded --timeout=60s
kubectl delete pod kafka-producer -n kafka

The function logs should show something like:

Received event
Context Attributes,
  specversion: 1.0
  type: dev.knative.kafka.event
  source: kafka://my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092/test-topic
  id: partition:0/offset:0
  time: 2026-07-16T10:10:21.59Z
  datacontenttype: application/octet-stream
Extensions,
  kafkaoffset: 0
  kafkapartition: 0
  kafkatopic: test-topic
Data (binary),
  Hello from Kafka!

Message with key

kubectl run kafka-producer-keyed -n kafka \
  --image=quay.io/strimzi/kafka:latest-kafka-4.2.0 \
  --restart=Never \
  --command -- sh -c \
  'echo "mykey:Hello with a key!" | bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic test-topic --property parse.key=true --property key.separator=:'

kubectl wait pod/kafka-producer-keyed -n kafka --for=jsonpath='{.status.phase}'=Succeeded --timeout=60s
kubectl delete pod kafka-producer-keyed -n kafka

The function logs should show something like:

Received event
Context Attributes,
  specversion: 1.0
  type: dev.knative.kafka.event
  source: kafka://my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092/test-topic
  subject: mykey
  id: partition:0/offset:1
  time: 2026-07-16T10:12:46.921Z
  datacontenttype: application/octet-stream
Extensions,
  kafkakey: mykey
  kafkaoffset: 1
  kafkapartition: 0
  kafkatopic: test-topic
Data (binary),
  Hello with a key!

CloudEvent message (pass-through)

Messages that already carry CloudEvents headers (ce_ prefix, per the
CloudEvents Kafka Protocol Binding binary content mode) are passed through
as-is without re-wrapping. The console producer supports headers via
--property parse.headers=true — input format is
header:value,header:value\t<message body> (tab-separated).

kubectl run kafka-ce-producer -n kafka \
  --image=quay.io/strimzi/kafka:latest-kafka-4.2.0 \
  --restart=Never \
  --command -- sh -c \
  'printf "ce_specversion:1.0,ce_type:com.example.test,ce_source:test-source,ce_id:test-id-123,content-type:application/json\t{\"greeting\":\"Hello CE!\"}" | bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic test-topic --property parse.headers=true'

kubectl wait pod/kafka-ce-producer -n kafka --for=jsonpath='{.status.phase}'=Succeeded --timeout=60s
kubectl delete pod kafka-ce-producer -n kafka

The function logs should show the original CE attributes passed through, not
the dev.knative.kafka.event wrapper:

Received event
Context Attributes,
  specversion: 1.0
  type: com.example.test
  source: test-source
  id: test-id-123
  datacontenttype: application/json
Data,
  {"greeting":"Hello CE!"}

Cleanup

kubectl delete ksvc my-kafka-func
kubectl delete kafka my-cluster -n kafka
kubectl delete kafkatopic test-topic -n kafka
kubectl delete -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka
kubectl delete namespace kafka
kind delete cluster --name kafka-test
rm -rf /tmp/my-kafka-func /tmp/kafka-func-build /tmp/func-local

knative-prow · 2026-07-16T11:57:52Z

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by: aliok
Once this PR has been reviewed and has the lgtm label, please assign lkingland for approval. For more information see the Code Review Process.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

Copilot

Pull request overview

Adds a new Kafka transport for Go Functions that consumes Kafka messages, converts them to CloudEvents, and invokes the same set of supported CloudEvents handler signatures used by the existing cloudevents transport. It also introduces a health-only HTTP server for readiness/liveness probes and exports receiver-function discovery so Kafka can reuse the CloudEvents signature support.

Changes:

Introduce kafka/ runtime package (service lifecycle, Sarama consumer group loop, Kafka→CloudEvents conversion, health endpoints).
Add unit tests for handler invocation signatures, health endpoints, env-var validation, and CloudEvents pass-through/wrapping behavior.
Export cloudevents.GetReceiverFn to reuse CloudEvents handler signature support from the Kafka runtime (and include a cmd/fkafka example).

Reviewed changes

Copilot reviewed 11 out of 12 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
kafka/service.go	Service lifecycle: starts consumer + health HTTP server; readiness/liveness endpoints; config reading.
kafka/service_test.go	Unit tests for readiness/liveness behavior.
kafka/message.go	Kafka message/header types used by the consumer and conversion logic.
kafka/instance.go	Handler signature validation and invocation across the supported CloudEvents signatures.
kafka/handler_test.go	Tests that all supported CloudEvents handler signatures are invokable via Kafka transport.
kafka/consumer.go	Sarama consumer group loop plus Kafka message → CloudEvents conversion (including CE header pass-through).
kafka/consumer_test.go	Tests for message wrapping, CE pass-through parsing, and missing-env-var validation.
cmd/fkafka/main.go	Example binary showing how Kafka transport can run a standard CloudEvents handler.
cloudevents/service.go	Switch to exported `GetReceiverFn` for handler lookup.
cloudevents/instance.go	Export `GetReceiverFn` so other transports (Kafka) can reuse signature support.
go.mod	Add Sarama and indirect deps needed for Kafka consumer support.
go.sum	Dependency checksum updates for the newly introduced Kafka/Sarama dependency tree.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot

Pull request overview

Copilot reviewed 11 out of 12 changed files in this pull request and generated 7 comments.

Copilot

Pull request overview

Copilot reviewed 11 out of 12 changed files in this pull request and generated 2 comments.

Copilot

Pull request overview

Copilot reviewed 11 out of 12 changed files in this pull request and generated 2 comments.

Copilot

Pull request overview

Copilot reviewed 11 out of 12 changed files in this pull request and generated 2 comments.

Copilot

Pull request overview

Copilot reviewed 11 out of 12 changed files in this pull request and generated 1 comment.

Add Kafka event sourcing middleware package

c9c80a2

knative-prow Bot requested review from matejvasek and matzew July 16, 2026 11:57

knative-prow Bot added the size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. label Jul 16, 2026

This was referenced Jul 16, 2026

[WIP] Add Kafka event sourcing for Go functions knative/func#3950

Open

Add Kafka middleware package #169

Closed

aliok added 6 commits July 16, 2026 16:55

Fix data race on responseEventWarned using sync.Once

922dacc

Parse ce_dataschema as standard CE attribute in pass-through

0d6421b

Handle *DefaultHandler pointer in invokeHandler

d50f078

Validate handler signature at startup instead of first message

608c28c

Move context.Context to first parameter in invokeHandler

15d7f26

Add unit tests for health handlers, CE pass-through, and consumeLoop

f06959a

aliok requested a review from Copilot July 16, 2026 13:56

Copilot started reviewing on behalf of aliok July 16, 2026 13:57 View session

Copilot AI reviewed Jul 16, 2026

View reviewed changes

Comment thread kafka/service.go

Comment thread kafka/consumer.go

Comment thread kafka/instance.go Outdated

aliok added 3 commits July 16, 2026 17:14

Cancel consumer context on shutdown so Sarama unblocks promptly

b9d48bd

Recover from GetReceiverFn panic in validateHandler

d035e92

Support content-type header for CE datacontenttype in Kafka pass-through

446dd5d

aliok requested a review from Copilot July 16, 2026 20:20

Copilot started reviewing on behalf of aliok July 16, 2026 20:21 View session

Copilot AI reviewed Jul 16, 2026

View reviewed changes

Comment thread kafka/service.go

Comment thread kafka/service.go

Comment thread kafka/service.go

Comment thread kafka/service.go

Comment thread kafka/service.go

Comment thread kafka/service.go

Comment thread cloudevents/instance.go

aliok added 2 commits July 16, 2026 23:36

Use buffered stop channel with non-blocking sends

8c6cded

Add GoDoc comment to exported GetReceiverFn

e50d677

aliok requested a review from Copilot July 16, 2026 20:48

Copilot started reviewing on behalf of aliok July 16, 2026 20:48 View session

Copilot AI reviewed Jul 16, 2026

View reviewed changes

Comment thread kafka/service.go

Comment thread kafka/service.go

aliok added 2 commits July 17, 2026 00:00

Close listener when startInstance fails to prevent port leak

6dfc999

Log debug message when static config file is absent

eb2e72d

aliok requested a review from Copilot July 16, 2026 21:07

Copilot started reviewing on behalf of aliok July 16, 2026 21:07 View session

Copilot AI reviewed Jul 16, 2026

View reviewed changes

Comment thread kafka/instance.go

Comment thread cloudevents/instance.go

Reject nil handler function in validateHandler

52ebda9

aliok requested a review from Copilot July 17, 2026 08:12

Copilot started reviewing on behalf of aliok July 17, 2026 08:12 View session

Copilot AI reviewed Jul 17, 2026

View reviewed changes

Comment thread kafka/consumer.go

Comment thread kafka/consumer.go

Set explicit Sarama Kafka version to V2_0_0_0

c446904

aliok requested a review from Copilot July 17, 2026 08:56

Copilot started reviewing on behalf of aliok July 17, 2026 08:56 View session

Copilot AI reviewed Jul 17, 2026

View reviewed changes

Comment thread kafka/instance.go

aliok added 2 commits July 17, 2026 12:09

Fix goimports and rename dummy to placeholder in tests

6121435

Fix missing newline at end of cmd/fkafka/main.go

ae16f93

Uh oh!

Conversation

aliok commented Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

How it works

Delivery semantics

Limitations

Testing Kafka Functions End-to-End on Kind

How it works

Limitations

Testing

Prerequisites

1. Build the func CLI

2. Create a Kind cluster with Knative

3. Install Kafka using Strimzi

4. Create a Kafka topic

5. Create the function

6. Add Kafka config

7. Build and deploy

Manual build (pre-release)

8. Deploy the Knative Service

9. Tail function logs

10. Send test messages

Plain message (wrapped as CloudEvent)

Message with key

CloudEvent message (pass-through)

Cleanup

Uh oh!

knative-prow Bot commented Jul 16, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

aliok commented Jul 16, 2026 •

edited

Loading